package tailLog

import (
	"code.oldboy.com/studygo/LogAgent/kafka"
	"context"
	"fmt"
	"github.com/hpcloud/tail"
	"time"
)


type TailTask struct {

	Path string
	Topic string
	tailFile *tail.Tail
	ctx context.Context
	cancelFunc context.CancelFunc
}

func (task *TailTask)Done(){
	task.cancelFunc()
}


func  NewTailTask(path string,topic string)(task * TailTask)  {
	ctx,cancelFunc:=context.WithCancel(context.Background())
	task=&TailTask{Path: path,Topic: topic,ctx: ctx,cancelFunc: cancelFunc}
	task.taskInit()
	return task
}

func (task *TailTask) taskInit(){
	config:=tail.Config{
		ReOpen: true,
		Follow: true,
		Location: &tail.SeekInfo{Offset: 0,Whence: 2},
		MustExist: false,
		Poll: true,
	}
	tailFile, _ := tail.TailFile(task.Path, config)
	task.tailFile=tailFile
	fmt.Println("init tail suceess")
	go task.run()
}





func (task *TailTask)readChannel() chan *tail.Line{
	return  task.tailFile.Lines
}


func (task *TailTask) run(){
	for{
		select {
		case line:=<-task.readChannel():
			fmt.Println("send")
			kafka.SendToChannel(task.Topic, line.Text)
		case <-task.ctx.Done():
			fmt.Printf("run collect exit topic:=%s path:=%s \n",task.Topic,task.Path	)
			return
		default:
			time.Sleep(time.Second)
		}
	}
}